package net.bw.realtime.tmall.dws.funcion1;

import net.bw.realtime.tmall.dwd.bean.TransactionEventBean;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import net.bw.realtime.tmall.common.utils.JdbcUtil;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @Author: liuyawei
 * @Date: 2025/5/27
 */

//todo b 实时订单量统计
public class ToOrderCntWindowFunction {
	public static void toOrderCnt(DataStream<TransactionEventBean> timeStream) {

		SingleOutputStreamOperator<String> order_cnt = timeStream
				.filter(x -> "place_order".equals(x.getActionType()))
				.keyBy(x -> "sku_id")
				.window(TumblingEventTimeWindows.of(Time.seconds(10)))
				.aggregate(new AggregateFunction<TransactionEventBean, Long, Long>() {

					@Override
					public Long createAccumulator() {
						return 0L;
					}

					@Override
					public Long add(TransactionEventBean TransactionEventBean, Long aLong) {
						return aLong + 1;
					}

					@Override
					public Long getResult(Long aLong) {

						return aLong;
					}

					@Override
					public Long merge(Long aLong, Long acc1) {
						return aLong + acc1;
					}
				}, new ProcessWindowFunction<Long, String, String, TimeWindow>() {
					private final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
					@Override
					public void process(String s, ProcessWindowFunction<Long, String, String, TimeWindow>.Context context, Iterable<Long> iterable, Collector<String> collector) throws Exception {
						Long next = iterable.iterator().next();
						String window_start = sdf.format(new Date(context.window().getStart()));
						String window_end = sdf.format(new Date(context.window().getEnd()));

						collector.collect(window_start + "," + window_end + "," + next + "," + System.currentTimeMillis());
					}
				})
				.name("order-count-aggregation")
				.uid("order-count-aggregation");

		//order_cnt.print();

		JdbcUtil.sinkToClickhouseUpsert(order_cnt,"INSERT INTO jtp_tmall_report.dws_order_count_window_report (\n" +
				"    window_start_time, window_end_time, order_count, ts\n" +
				") VALUES (\n" +
				"             ?, ?, ?, ?\n" +
				"         )");
	}


}
